package com.zp.qualitypractice.rx;

import android.util.Log;

import com.zp.qualitypractice.ALog;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

/**
 * Created by izp on 2017/9/1.
 */

public class RxBeginCode {
    private static final String TAG = "RxBegin";
    /**
     * 用上下游来描述Rx的事件流订阅关系
     */
    private void upStreamAndDownStream(){
//创建一个上游 Observable：
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                用emitter发射器连续发送事件。
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        });
        //创建一个下游 Observer
//        observer会连续接收事件，直到onComplete();事件
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                ALog.v(TAG,"subscribe");
            }

            @Override
            public void onNext(Integer value) {
                ALog.v(TAG, "" + value);
            }

            @Override
            public void onError(Throwable e) {
                ALog.v(TAG, "error");
            }

            @Override
            public void onComplete() {
                ALog.v(TAG, "complete");
            }
        };
        //建立连接
        observable.subscribe(observer);
//subscribe调用时，整个上下游的程序才运行
    }

    /**
     * 上下游结合起来就是平常使用Rx进行网络请求的状态。RxJava引以为傲的链式操作
     */
    private void combineUpStreamDownStream(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                ALog.v(TAG, "subscribe");
            }

            @Override
            public void onNext(Integer value) {
                ALog.v(TAG, "" + value);
            }

            @Override
            public void onError(Throwable e) {
                ALog.v(TAG, "error");
            }

            @Override
            public void onComplete() {
                ALog.v(TAG, "complete");
            }
        });
    }

    public void leanDisposable(){
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                ALog.v(TAG, "emit 1");
                emitter.onNext(1);
                ALog.v(TAG, "emit 2");
                emitter.onNext(2);
                ALog.v(TAG, "emit 3");
                emitter.onNext(3);
                ALog.v(TAG, "emit complete");
                emitter.onComplete();
                ALog.v(TAG, "emit 4");
                emitter.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            private Disposable mDisposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                ALog.v(TAG, "subscribe");
                mDisposable = d;
            }

            @Override
            public void onNext(Integer value) {
                ALog.v(TAG, "onNext: " + value);
                i++;
                if (i == 2) {
                    ALog.v(TAG, "dispose");
                    mDisposable.dispose();
                    ALog.v(TAG, "isDisposed : " + mDisposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                ALog.v(TAG, "error");
            }

            @Override
            public void onComplete() {
                ALog.v(TAG, "complete");
            }
        });
    }
}
